package com.raymond.queue.handler;

import com.alibaba.fastjson.JSONObject;
import com.raymond.queue.FileQueue;
import com.raymond.queue.QueueManager;
import com.raymond.queue.common.msg.request.CreateTopicRequest;
import com.raymond.queue.common.msg.response.CreateTopicResponse;
import com.raymond.queue.utils.FileQueueBuilder;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.internal.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.RandomAccessFile;
import java.net.InetSocketAddress;
import java.nio.channels.FileChannel;


/**
 * 创建topic
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2021-07-14 14:39
 */
public class CreateTopicRequestHandler extends SimpleChannelInboundHandler<CreateTopicRequest> {

    private final static Logger logger = LoggerFactory.getLogger(CreateTopicRequestHandler.class);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, CreateTopicRequest msg) {
        logger.info("创建topic:{}", JSONObject.toJSONString(msg));
        if(QueueManager.getFileQueue(msg.getTopic()) != null) {
            ctx.writeAndFlush(new CreateTopicResponse(0, msg.getSequenceId()));
            return;
        }

        FileQueueBuilder<byte[]> byteFileQueueBuilder = FileQueueBuilder.create(byte[].class, msg.getTopic());
        if (!StringUtil.isNullOrEmpty(msg.getGroupName())) {
            byteFileQueueBuilder.setGroupName(msg.getGroupName());
        }
        byteFileQueueBuilder.setQueueModel(msg.getQueueModel() == 0 ? FileQueue.QueueModel.ORDINARY : FileQueue.QueueModel.SUBSCRIBE);
        try {
            FileQueue<byte[]> build = byteFileQueueBuilder.setGrowMode(FileQueue.GrowMode.getGrowMode(msg.getGrowMode())).build();
            if (build == null) {
                ctx.writeAndFlush(new CreateTopicResponse(1, msg.getSequenceId()));
            } else {
                QueueManager.put(build);
                ctx.writeAndFlush(new CreateTopicResponse(0, msg.getSequenceId()));
            }
        } catch (Exception e) {
            ctx.writeAndFlush(new CreateTopicResponse(1, msg.getSequenceId()));
            logger.error("创建topic异常,topic:{},原地址:{}", msg.getTopic(), ctx.channel().remoteAddress(),e);
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("CreateTopicRequestHandler Established connection with the remote client.");

        // do something

        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("CreateTopicRequestHandler Disconnected with the remote client.");

        // do something

        ctx.fireChannelInactive();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}
